package p.ithorns.framework.common.thread;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nonnull;
import java.util.Collection;
import java.util.Set;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

/**
 * ExpiredCache
 *
 * @author ithorns
 * @version 1.0
 * @date 2020/8/18 9:05
 */
public class ExpiredCache<K, V> implements Runnable {

    private static final Logger log = LoggerFactory.getLogger(ExpiredCache.class);

    private final static ExecutorService EXPIRE_NOTIFY_EXECUTOR = ThreadPoolFactory.create(
            ThreadPoolFactory.SYSTEM_CORE_SIZE,
            ThreadPoolFactory.SYSTEM_CORE_SIZE * 2,
            "expired-cache"
    );

    private final static int DEFAULT_CAPACITY = 128;

    private final ConcurrentMap<K, V> cahce;

    private final DelayQueue<CacheItem<K>> queue;

    private final Thread daemonThread;

    private final ExpirationListener<K, V> expirationListener;

    private ExpiredCache(String cacheName,
                         int capacity,
                         ExpirationListener<K, V> expirationListener) {
        this.cahce = new ConcurrentHashMap<>(capacity);
        this.queue = new DelayQueue<>();
        this.expirationListener = expirationListener;

        daemonThread = ThreadPoolFactory
                .threadFactory(cacheName, false)
                .newThread(this);
        daemonThread.setDaemon(true);
        daemonThread.setName(cacheName);
        daemonThread.start();
    }

    public void put(K key, V value, long time, TimeUnit unit) {
        synchronized (this) {
            V oldValue = cahce.put(key, value);
            if (oldValue != null) {
                queue.removeIf(e -> e.key.equals(key));
            }
            long nanoTime = TimeUnit.NANOSECONDS.convert(time, unit);
            queue.put(new CacheItem<>(key, nanoTime));
        }
    }

    public V get(K key) {
        return cahce.get(key);
    }

    @SuppressWarnings("InfiniteLoopStatement")
    @Override
    public void run() {
        for (; ; ) {
            try {
                CacheItem<K> delayed = queue.take();
                // 超时对象处理
                K k = delayed.getKey();
                V v = cahce.get(k);
                cahce.remove(k);
                if (expirationListener != null) {
                    // 异步
                    //if (EXPIRE_NOTIFY_EXECUTOR != null) {
                    EXPIRE_NOTIFY_EXECUTOR.execute(() ->
                            expirationListener.onExpired(k, v)
                    );
                    // }
                    // !!! 同步通知 会阻塞，导致时间不准
                    //else {
                    //    expirationListener.onExpired(k, v);
                    //}
                }
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
        }
    }

    public void remove(K key) {
        synchronized (this) {
            cahce.remove(key);
            queue.removeIf(e -> e.key.equals(key));
        }
    }

    public Set<K> keySet() {
        synchronized (cahce) {
            return new CopyOnWriteArraySet<>(cahce.keySet());
        }
    }

    public Collection<V> values() {
        synchronized (cahce) {
            return new CopyOnWriteArrayList<>(cahce.values());
        }
    }

    public boolean containsKey(K key) {
        return cahce.containsKey(key);
    }

    public void destroy() {
        if (daemonThread.isAlive()) {
            daemonThread.interrupt();
        }

        if (null != EXPIRE_NOTIFY_EXECUTOR) {
            EXPIRE_NOTIFY_EXECUTOR.shutdown();
        }
    }

    static class CacheItem<K> implements Delayed {

        private static final long NANO_ORIGIN = System.nanoTime();

        private static final AtomicLong SEQ = new AtomicLong(0);

        private final long sequenceNumber;

        private final long time;

        private final K key;

        public CacheItem(K submit, long timeout) {
            this.time = now() + timeout;
            this.key = submit;
            this.sequenceNumber = SEQ.getAndIncrement();
        }

        private static long now() {
            return System.nanoTime() - NANO_ORIGIN;
        }

        public K getKey() {
            return key;
        }

        @Override
        public long getDelay(TimeUnit unit) {
            return unit.convert(time - now(), TimeUnit.NANOSECONDS);
        }

        @Override
        public int compareTo(@Nonnull Delayed other) {
            // compare zero ONLY if same object
            if (other == this) {
                return 0;
            }

            if (other instanceof ExpiredCache.CacheItem) {
                CacheItem x = (CacheItem) other;
                long diff = time - x.time;
                if (diff < 0) {
                    return -1;
                } else if (diff > 0) {
                    return 1;
                } else if (sequenceNumber < x.sequenceNumber) {
                    return -1;
                } else {
                    return 1;
                }
            }

            long d = (getDelay(TimeUnit.NANOSECONDS) - other.getDelay(TimeUnit.NANOSECONDS));
            return (d == 0) ? 0 : ((d < 0) ? -1 : 1);
        }
    }

    public static <K, V> ExpiredCache.Builder<K, V> newBuilder() {
        return new ExpiredCache.Builder<>();
    }


    /**
     * LocalExpiredCache构建器
     */
    public static final class Builder<K, V> {
        /**
         * 监听器
         */
        private ExpirationListener<K, V> expirationListener;
        /**
         * 缓存的名称
         */
        private String cacheName = "default";

        /**
         * 缓存大小
         */
        private int capacity = DEFAULT_CAPACITY;

        private Builder() {
        }

        public Builder<K, V> capacity(int capacity) {
            this.capacity = capacity;
            return this;
        }

        /**
         * 配置过期监听器
         *
         * @param expirationListener 过期监听器(包含过期后触发的操作)
         * @return Builder
         */
        public Builder<K, V> expirationListener(ExpirationListener<K, V> expirationListener) {
            this.expirationListener = expirationListener;
            return this;
        }

        /**
         * 配置缓存队列名称
         *
         * @param cacheName 缓存队列名称
         * @return Builder
         */
        public Builder<K, V> cacheName(String cacheName) {
            this.cacheName = cacheName == null ? "default" : cacheName;
            return this;
        }

        /**
         * 构造结构对象
         *
         * @return ExpiredCache
         */
        public ExpiredCache<K, V> build() {
            return new ExpiredCache<>(
                    this.cacheName,
                    this.capacity,
                    this.expirationListener
            );
        }
    }

}
